package kafka

import (
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"github.com/zeromicro/go-zero/core/logx"
	"go.uber.org/zap"
)

type Producer struct {
	producer sarama.AsyncProducer
}

func NewProducer(config *sarama.Config, addrs []string) *Producer {
	producer, err := sarama.NewAsyncProducer(addrs, config)
	if err != nil {
		panic("kafka producer start err")
	}
	go handleSuccess(producer)
	go handleError(producer)
	return &Producer{producer: producer}
}

func NewDefaultProducer(addrs []string) *Producer {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForLocal
	config.Producer.Partitioner = sarama.NewHashPartitioner
	// 发送成功时,通知信息
	config.Producer.Return.Successes = true
	// 发送失败时,通知信息
	config.Producer.Return.Errors = true
	return NewProducer(config, addrs)
}

func handleSuccess(producer sarama.AsyncProducer) {
	var (
		pm *sarama.ProducerMessage
	)
	for {
		pm = <-producer.Successes()
		if pm != nil {
			logx.Info(fmt.Sprintf("producer message success, partition:%d offset:%d key:%v valus:%s", pm.Partition, pm.Offset, pm.Key, pm.Value))
			fmt.Printf(fmt.Sprintf("producer message success, partition:%d offset:%d key:%v valus:%s\n", pm.Partition, pm.Offset, pm.Key, pm.Value))
		}
	}
}

func handleError(producer sarama.AsyncProducer) {
	var (
		err *sarama.ProducerError
	)
	for {
		err = <-producer.Errors()
		if err != nil {
			logx.Error(fmt.Sprintf("producer message error, partition:%d offset:%d key:%v valus:%s error(%v)", err.Msg.Partition, err.Msg.Offset, err.Msg.Key, err.Msg.Value, err), zap.Error(err))
		}
	}
}

// PutMsg put msg to kafka
func (p *Producer) PutMsg(packet interface{}, topic string) error {
	var (
		vBytes []byte
		err    error
	)
	if vBytes, err = json.Marshal(packet); err != nil {
		//log.Get().Error("put msg is not json")
		return err
	}
	p.producer.Input() <- &sarama.ProducerMessage{Topic: topic, Value: sarama.ByteEncoder(vBytes)}
	return nil
}
